package com.example.window;

import com.example.model.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Created with IntelliJ IDEA.
 * ClassName: StreamWindow
 * Package: com.example.window
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-23
 * Time: 12:55
 */


//窗口 对无限流数据分割成有限的数据块
public class StreamWindowAll {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> map = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {

                        final String[] s = value.split(" ");

                        return new WaterSensor(s[0], Long.valueOf(s[1]), Integer.valueOf(s[2]));
                    }
                });

        //如果没有经过KeyBy 只能调用windowAll
        //windowAll 参数是 WindowAssigner 窗口分配器
        //TumblingEventTimeWindows 系统封装好的 窗口分配器规格
        //对于没有经过KeyBy操作的窗口来说 窗口逻辑就在一个任务上执行 并行度为1
        //而且手动调用窗口算子大小的并行度也是无效的 windowAll本身就是一个非并行的操作
        map.windowAll(TumblingEventTimeWindows.of(Time.seconds(2)));

        env.execute();
    }

}
